package ca.ericslandry.server.dao;

import java.util.Collections;
import java.util.Random;

import javax.cache.Cache;
import javax.cache.CacheException;
import javax.cache.CacheManager;

import ca.ericslandry.shared.domain.DatastoreCounter;
import ca.ericslandry.shared.domain.DatastoreCounterShard;

import com.googlecode.objectify.ObjectifyService;
import com.googlecode.objectify.Query;
import com.googlecode.objectify.helper.DAOBase;

/**
 * A counter which can be incremented rapidly.
 * 
 * Capable of incrementing the counter and increasing the number of shards. When
 * incrementing, a random shard is selected to prevent a single shard from being
 * written to too frequently. If increments are being made too quickly, increase
 * the number of shards to divide the load. Performs datastore operations using
 * JDO.
 * 
 * Lookups are attempted using Memcache (Jcache). If the counter value is not in
 * the cache, the shards are read from the datastore and accumulated to
 * reconstruct the current count.
 * 
 * @author j.s@google.com (Jeff Scudder)
 * @author <a href="mailto:eric.s.landry@gmail.com">Eric Landry</a>
 *         (Objectified)
 */
public class ShardedCounter extends DAOBase {
	static {
		ObjectifyService.register(DatastoreCounter.class);
		ObjectifyService.register(DatastoreCounterShard.class);
	}
	private String counterName;
	private Cache cache;

	public ShardedCounter(String counterName) {
		this.counterName = counterName;
		cache = null;
		try {
			cache = CacheManager.getInstance().getCacheFactory().createCache(Collections.emptyMap());
		} catch (CacheException e) {
			// It's ok if we can't initialize our cache object since we can fall
			// back to querying the datastore.
		}

		if (!isInDatastore()) {
			DatastoreCounter entity = new DatastoreCounter(counterName, 1);
			ofy().put(entity);
		}
	}

	public String getCounterName() {
		return counterName;
	}

	private DatastoreCounter getThisCounter() {
		return ofy().query(DatastoreCounter.class).filter("counterName", counterName).get();
	}

	public boolean isInDatastore() {
		return getThisCounter() != null;
	}

	@SuppressWarnings("unchecked")
	public int getCount() {
		if (cache != null) {
			Integer cachedCount = (Integer) cache.get("count" + counterName);
			if (cachedCount != null) {
				return cachedCount;
			}
		}

		int sum = 0;
		Query<DatastoreCounterShard> q = ofy().query(DatastoreCounterShard.class).filter("counterName = ", counterName);
		for (DatastoreCounterShard shard : q) {
			sum += shard.getCount();
		}

		// Update the cache with the count we just found.
		if (cache != null) {
			cache.put("count" + counterName, new Integer(sum));
		}

		return sum;
	}

	@SuppressWarnings("unchecked")
	public int getNumShards() {
		if (cache != null) {
			Integer cachedCount = (Integer) cache.get("shards" + counterName);
			if (cachedCount != null) {
				return cachedCount;
			}
		}

		int numShards = 0;
		// Find the current number of shards for this DatastoreCounter.
		DatastoreCounter current = getThisCounter();
		if (current != null) {
			numShards = current.getNumShards();
		}

		// Update the cached value for the number of shards.
		if (cache != null) {
			cache.put("shards" + counterName, new Integer(numShards));
		}

		return numShards;
	}

	public int addShard() {
		return addShards(1);
	}

	@SuppressWarnings("unchecked")
	public int addShards(int count) {
		int numShards = 0;
		// Find the initial shard count for this Counter.
		DatastoreCounter current = getThisCounter();
		if (current != null) {
			numShards = current.getNumShards();
			current.setNumShards(numShards + count);
			ofy().put(current);
		}

		// Create new shard objects for this counter.
		for (int i = 0; i < count; i++) {
			DatastoreCounterShard newShard = new DatastoreCounterShard(getCounterName(), numShards);
			ofy().put(newShard);
			numShards++;
		}

		if (cache != null) {
			cache.put("shards" + counterName, new Integer(numShards));
		}

		return numShards;
	}

	public void increment() {
		increment(1);
	}

	@SuppressWarnings("unchecked")
	public void increment(int count) {
		if (cache != null) {
			Integer cachedCount = (Integer) cache.get("count" + counterName);
			if (cachedCount != null) {
				cache.put("count" + counterName, new Integer(count + cachedCount.intValue()));
			}
		}

		// Find how many shards are in this counter.
		int shardCount = 0;
		DatastoreCounter current = getThisCounter();
		shardCount = current.getNumShards();

		// Choose the shard randomly from the available shards.
		Random generator = new Random();
		int shardNum = generator.nextInt(shardCount);

		DatastoreCounterShard shard = ofy().query(DatastoreCounterShard.class).filter("counterName = ", counterName)
				.filter("shardNumber", shardNum).get();
		if (shard != null) {
			shard.increment(count);
			ofy().put(shard);
		}
	}
}
